package baseFx.common.mq.rabbitmq;

import baseFx.common.EventDispatcher;
import baseFx.common.concurrent.Task;
import baseFx.common.mq.MqMessage;
import baseFx.common.mq.MqSubscriber;
import baseFx.common.mq.SubscriberError;
import baseFx.common.utils.ExceptionUtils;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeoutException;

public class RabbitMqSubscriber implements MqSubscriber {
    private static Logger logger = LoggerFactory.getLogger(RabbitMqSubscriber.class);
    private static boolean isShutdown;
    private static String receivedKey = UUID.randomUUID().toString();
    private static String errorKey = UUID.randomUUID().toString();
    private EventDispatcher<MqMessage> receivedDispatcher = new EventDispatcher<>(receivedKey);
    private EventDispatcher<SubscriberError> errorDispatcher = new EventDispatcher<>(errorKey);
    private RabbitMqContext context;
    private Map<String, List<DefaultConsumer>> consumerMap = new HashMap<>();
    private Map<String, Connection> connMap = new HashMap<>();
    private boolean closing = false;
    private String id = UUID.randomUUID().toString();

    static {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> isShutdown = true));
    }

    /**
     * 本地缓存数量
     */
    public int prefetchCount = 200;

    public RabbitMqSubscriber(RabbitMqContext context) {
        this.context = context;
    }

    @Override
    public void declare(String queue) {
        if (context.existsExchange(queue) == false) {
            context.exchangeDeclare(queue, BuiltinExchangeType.FANOUT);
        }
    }

    @Override
    public EventDispatcher<MqMessage> eventReceived() {
        return receivedDispatcher;
    }

    @Override
    public EventDispatcher<SubscriberError> eventError() {
        return errorDispatcher;
    }

    protected void onReceived(MqMessage message) {
        receivedDispatcher.fire(receivedKey, this, message);
    }

    @Override
    public void consume(String queue, String name, Object state) {
        if (closing || isShutdown) {
            return;
        }
        try {
            if (context.existsQueue(name) == false) {
                context.queueDeclare(name, true);
            }
            context.queueBind(queue, name);
            String key = queue + "|" + name;
            if (consumerMap.containsKey(key) == false) {
                consumerMap.put(key, new ArrayList<>());
            }
            if (connMap.containsKey(key) == false) {
                connMap.put(key, context.getConnection());
            }
            String tag = UUID.randomUUID().toString();
            Channel channel = connMap.get(key).createChannel();
            channel.basicQos(prefetchCount);
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                private Object consumerState = state;
                private String consumerEx = queue;
                private String consumerQueue = name;
                private String consumerTag = tag;
                private Channel consumerChannel = channel;
                private String consumerKey = key;
                private boolean isClosed;

                @Override
                public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
                    if (isShutdown) {
                        return;
                    }
                    logger.warn("消费者异常", sig);
                    SubscriberError args = new SubscriberError(sig, consumerState);
                    errorDispatcher.fire(errorKey, RabbitMqSubscriber.this, args);
                    if (args.isRetry == false) {
                        return;
                    }
                    while (isShutdown == false) {
                        Task.delay(1000, 5000);
                        try {
                            consume(consumerEx, consumerQueue, consumerState);
                            break;
                        } catch (Exception ex) {
                            logger.warn("重启失败", ex);
                            SubscriberError exArgs = new SubscriberError(ex, consumerState);
                            errorDispatcher.fire(errorKey, RabbitMqSubscriber.this, exArgs);
                            if (exArgs.isRetry == false) {
                                throw ex;
                            }
                        }
                    }
                }

                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    if (isClosed || isShutdown) {
                        return;
                    }
                    MqMessage msg = new MqMessage();
                    msg.queue = consumerEx;
                    msg.name = consumerQueue;
                    msg.data = new String(body, StandardCharsets.UTF_8);
                    msg.state = consumerState;
                    if (logger.isDebugEnabled()) {
                        logger.debug("{}-{}：{}", msg.queue, msg.name, msg.data);
                    }
                    try {
                        RabbitMqSubscriber.this.onReceived(msg);
                        if (consumerChannel.isOpen()) {
                            if (msg.reconsume) {
                                consumerChannel.basicReject(envelope.getDeliveryTag(), true);
                            } else {
                                consumerChannel.basicAck(envelope.getDeliveryTag(), false);
                            }
                        }
                    } catch (Exception e) {
                        logger.warn("处理异常：{}-{}", msg.queue, msg.name, e);
                        try {
                            isClosed = true;
                            consumerMap.get(consumerKey).remove(this);
                            consumerChannel.basicCancel(consumerTag);
                            consumerChannel.close();
                        } catch (IOException | TimeoutException ex) {
                            ex.printStackTrace();
                        }
                    }
                }
            };
            channel.basicConsume(name, false, tag, consumer);
            consumerMap.get(key).add(consumer);
        } catch (IOException e) {
            cancel(queue, name);
            throw ExceptionUtils.toRuntime(e);
        }
    }

    @Override
    public void cancel(String queue, String name) {
        cancel(queue + "|" + name);
    }

    private void cancel(String key) {
        List<DefaultConsumer> list = consumerMap.remove(key);
        if (list == null) {
            return;
        }
        try {
            for (DefaultConsumer item : list) {
                item.getChannel().basicCancel(item.getConsumerTag());
                item.getChannel().close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() {
        closing = true;
        if (consumerMap != null) {
            for (String key : new ArrayList<>(consumerMap.keySet())) {
                cancel(key);
            }
            consumerMap = null;
        }
        if (receivedDispatcher != null) {
            receivedDispatcher.clear();
            receivedDispatcher = null;
        }
        context = null;
        closing = false;
    }
}
